Deavmi's coding shack
Commit cc973dfd54fff4f2d32eb71135e8d5bd3d1f246f
Parents : cd28398
Author : Tristan Brice Velloza Kildaire <deavmi@redxen.eu>
Date : 2026-05-03T19:40:06+02:00
properly abstracted and did the hole "PipeMonitor" now handles its acquisition
Changes
Diff
diff --git a/monitor.go b/monitor.go
index f33bb7f..185584b 100644
--- a/monitor.go
+++ b/monitor.go
@@ -31,7 +31,7 @@ type Bruh struct {
cmd *exec.Cmd
// pipes monitored
- pipes []io.ReadCloser
+ pipes []*PipeMonitor
pipesLock sync.Mutex
// pipes []io.Closer
@@ -104,7 +104,8 @@ func (b *Bruh) Start() {
// TODO: Handle this
b.log.Error("Could not obtain standard output pipe for '%s': %v", b.u.Name(), err)
} else {
- b.pipes = append(b.pipes, std_out)
+ // create a new pipe minitor and start it up
+ NewPipeMonitor(b, std_out, b.ctx).Start()
}
std_err, err := e.StderrPipe()
@@ -112,12 +113,19 @@ func (b *Bruh) Start() {
// TODO: Handle this
b.log.Error("Could not obtain standard error pipe for '%s': %v", b.u.Name(), err)
} else {
- b.pipes = append(b.pipes, std_err)
+ // create a new pipe minitor and start it up
+ NewPipeMonitor(b, std_err, b.ctx).Start()
}
+}
- go b.logMonitor(std_out, b.ctx)
- go b.logMonitor(std_err, b.ctx)
+func (b *Bruh) AddPipeMon(pipeMon *PipeMonitor) {
+ b.pipesLock.Lock()
+ b.pipes = append(b.pipes, pipeMon)
+ b.pipesLock.Unlock()
+}
+func NewPipeMonitor(monitor *Bruh, r io.ReadCloser, ctx context.Context) PipeMonitor {
+ return PipeMonitor{m: monitor, pipe: r, ctx: ctx}
}
func (b *Bruh) monitor() {
@@ -131,7 +139,7 @@ func (b *Bruh) monitor() {
b.log.Trace("Shutting down pipes...")
for p_idx, p := range b.pipes {
b.log.Trace("(%d) Shutting down pipe '%v'...", p_idx, p)
- p.Close()
+ p.Stop()
}
}
@@ -147,6 +155,9 @@ type PipeMonitor struct {
// pipe
pipe io.ReadCloser
+ // context
+ ctx context.Context
+
// logging
log *stdlogger.StdLogger
}
@@ -155,23 +166,44 @@ func (pm PipeMonitor) String() string {
return fmt.Sprintf("%T (unit: %s, pipe: %v)", pm, pm.m.Name(), pm.pipe)
}
+// Spawns a go routine to start this PipeMonitor
+//
+// This will eventually cause it to be added to
+// the Monitor's pipe list
+func (pm PipeMonitor) Start() {
+ go pm.logMonitor()
+}
+
+// Stops the pipe monmitor
+//
+// This is accomplished by:
+//
+// 1. Updating the context (TODO: not needed but it is nice)
+// 2. Close()'ing the ReadCloser calling the PipeMonitor's
+// `Read()` to unblock and the Goroutine to end
+func (pm PipeMonitor) Stop() {
+ pm.log.Debug("Stopping %v...", pm)
+ if e := pm.pipe.Close(); e != nil {
+ pm.log.Error("Error when closing internal pipe during shutdown: %v", e)
+ }
+ pm.log.Debug("%v stopped", pm)
+}
+
// log monitor for the given reader
//
// waits for bytes and when it gets
// an error then it exits.
-func (b *Bruh) logMonitor(r io.ReadCloser, ctx context.Context) {
+func (b *PipeMonitor) logMonitor() {
// Add `r` to b.pipes
- b.pipesLock.Lock()
- b.pipes = append(b.pipes, r)
- b.pipesLock.Unlock()
+ b.m.AddPipeMon(b)
- b.log.Trace("Log monitor for '%s' on fd '%v' enter", b.u.Name(), r)
+ b.log.Trace("Log monitor for '%s' on fd '%v' enter", b.m.Name(), b.pipe)
b.log.Trace("Log monitor for '%s' on fd '%v' using buffer size %d", BUFFER_SIZE)
var buff []byte = make([]byte, BUFFER_SIZE)
for {
// we read at most `BUFFER_SIZE` bytes per call
- n, e := r.Read(buff)
+ n, e := b.pipe.Read(buff)
if e != nil {
// TODO: Check context in case maybe something basd happened
// like process closed one end of pipe _instead_ of us initiating
Served by rngit 1.5.0 - Generated in 0.05s